package com.zfei.firstflink;

import com.zfei.firstflink.model.XexHomeOrder;
import com.zfei.firstflink.utils.MysqlSource;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;

import org.apache.flink.api.java.tuple.Tuple0;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.api.java.Tumble;
import org.apache.flink.types.Row;

/***
 *$com.zfei.firstflink.TestMysqlSource
 *@author:jiwenfei
 *
 *@version:
 *
 *createTime:2018/8/17:上午10:40
 *
 *description:
 */
public class TestMysqlSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
        XexHomeOrder order=new XexHomeOrder();
        order.setOrderNo("1213223");
        order.setState(1);
        //order.setPrice(812322.23);

        XexHomeOrder order1=new XexHomeOrder();
        order1.setOrderNo("31213223");
        order1.setState(1);
       // order1.setPrice(3812.23);

        XexHomeOrder order2=new XexHomeOrder();
        order2.setOrderNo("4213223");
        order2.setState(2);
        //order2.setPrice(4123.23);
        DataStream<XexHomeOrder> orderStream =
                //env.fromElements(order,order1,order2);
                env.addSource(   new MysqlSource()).filter(new FilterFunction<XexHomeOrder>() {
                    @Override
                    public boolean filter(XexHomeOrder xexHomeOrder) throws Exception {
                        if (xexHomeOrder.getState()==null||xexHomeOrder.getOrderTime()==null){
                            return  false;
                        }else
                            return true;
//                        item->item.getState()==null||item.getOrderNo()==null
//                        return false;
                    }
                });
//        orderStream.map(new MapFunction<XexHomeOrder, Tuple2<Integer,Long>>() {
//            @Override
//            public Tuple2<Integer, Long> map(XexHomeOrder xexHomeOrder) {
//                return new Tuple2(xexHomeOrder.getOrderType(),1L);
//            }
//        }).keyBy(1).sum(1);
                //.print();

//        orderStream.map(new MapFunction<XexHomeOrder, Tuple1<Integer>>() {
//            @Override
//            public Tuple1<Integer> map(XexHomeOrder xexHomeOrder) {
//                return new Tuple1<>(1);
//            }
//        }).keyBy(0).reduce((t,t1)->{
//            return new Tuple1<Integer>(t.f0+1);
//        }).print();



//        orderStream.filter(t->(t.getState()==8)).map(new MapFunction<XexHomeOrder, Tuple1<Integer>>() {
//            @Override
//            public Tuple1<Integer> map(XexHomeOrder xexHomeOrder) {
//                return new Tuple1<>(1);
//            }
//        }).keyBy(0).sum(0)
//                .reduce((t,t1)->{
//            return new Tuple1<Integer>(t.f0+1);
//        })
//                .print();

        StreamTableEnvironment sTableEnv = TableEnvironment.getTableEnvironment(env);
       sTableEnv.registerDataStream("orderStream",orderStream,"state as sta, orderType as orderType, price as priceT,proctime.proctime");
        Table sqlResult1  =
                sTableEnv.scan("orderStream");//.filter("orderType==4").groupBy("sta")
        Table sqlResult2=sqlResult1.where("orderType==4")
//                .select("order ")
                .window(Tumble.over("2.days").on("proctime").as("dayWindow"))
                .groupBy("dayWindow,sta")
                .select("sta,priceT.avg as avgBill,dayWindow.end ")
                ;

        sTableEnv.toRetractStream( sqlResult2,Row.class).map(item->{

          System.out.println(item.f1.toString());
            return "12";
        });


//        Table table= sTableEnv.fromDataStream(orderStream);
//        sTableEnv.registerDataStream("orderStream",orderStream);
//        sTableEnv.toAppendStream(sTableEnv.scan("orderStream").filter("f0.state!=8").select("f0.orderNo,f0.state"),XexHomeOrder.class).map(item->{
//            System.out.println(item.getOrderNo()+"_"+item.getState());
//            return 1;
//        });
        env.execute("ee");
    }
    public  static  class  DDD {
        public String on;
        public Integer sta;

        public String getOn() {
            return on;
        }

        public void setOn(String on) {
            this.on = on;
        }

        public Integer getSta() {
            return sta;
        }

        public void setSta(Integer sta) {
            this.sta = sta;
        }

        public DDD(){

        }
    }
}
